package com.ahcloud.edas.common.kafka.client;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Properties;

/**
 * @program: edas-common
 * @description:
 * @author: YuKai Fan
 * @create: 2024/4/19 21:10
 **/
public class KafkaMessageClient {

    private final String env;
    private final String clientId;
    private final List<String> bootstrapServers;
    private final Properties producerProperties = new Properties();
    private final Properties consumerProperties = new Properties();


    public KafkaMessageClient(String env,
                              String clientId,
                              List<String> bootstrapServers,
                              Map<String, Object> producerProperties,
                              Map<String, Object> consumerProperties) {
        this.env = env;
        this.clientId = clientId;
        this.producerProperties.putAll(producerProperties);
        this.consumerProperties.putAll(consumerProperties);
        this.bootstrapServers = bootstrapServers;
    }

    /**
     * 构建生产者
     *
     * @param <K>
     * @param <V>
     * @return
     */
    public <K, V> KafkaProducer<K, V> createProducer() {
        return this.createProducer(null);
    }

    /**
     * 构建生产者
     *
     * @param <K>
     * @param <V>
     * @return
     */
    public <K, V> KafkaProducer<K, V> createProducer(Properties properties) {
        buildCommonProperties();
        if (Objects.nonNull(properties)) {
            this.producerProperties.putAll(properties);
        }
        return new KafkaProducer<>(this.producerProperties);
    }


    /**
     * 构建消费者
     *
     * @param <K>
     * @param <V>
     * @return
     */
    public <K, V> KafkaConsumer<K, V> createConsumer() {
        return this.createConsumer(null);
    }

    /**
     * 构建生产者
     *
     * @param <K>
     * @param <V>
     * @return
     */
    public <K, V> KafkaConsumer<K, V> createConsumer(Properties properties) {
        buildCommonProperties();
        if (Objects.nonNull(properties)) {
            this.consumerProperties.putAll(properties);
        }
        return new KafkaConsumer<>(this.consumerProperties);
    }

    public String getEnv() {
        return env;
    }

    public Properties getProducerProperties() {
        return producerProperties;
    }

    public Properties getConsumerProperties() {
        return consumerProperties;
    }

    private void buildCommonProperties() {
        if (!this.bootstrapServers.isEmpty()) {
            this.producerProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.bootstrapServers);
            this.consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.bootstrapServers);
        }
        if (this.clientId != null) {
            this.producerProperties.put(ProducerConfig.CLIENT_ID_CONFIG, this.clientId);
            this.consumerProperties.put(ConsumerConfig.CLIENT_ID_CONFIG, this.clientId);
        }
    }
}
